/** Easy parallelize job scheme.
 *
 *  Easy parallelize job scheme.
 *
 *  There are many group in this job, each group is form by some remote process
 *  and each in one group run in the same throw path.
 *
 *  @author : huicong
 *  @date : 2008-11-03
 *
 */

#include "EasyParallelJob.h"
#include <iostream>
#include <sstream>

namespace bcmi_frame
{

    EasyParallelJob::EasyParallelJob()
    {
    }

    EasyParallelJob::~EasyParallelJob()
    {
        delete _groupManager;
    }

    VOID EasyParallelJob::linkFrame(_IN ParallelFrame *_frame)
    {
        _myFrame = _frame;
        _groupManager = new EasyParallelGroupManager(_frame->getMPIProcessNumber());
        _requestBufferSize = _myFrame->getMPIProcessNumber() + sizeof(UINT) * 2;
    }

    VOID EasyParallelJob::insertGroup(IN UINT _pid, IN UINT _gid)
    {
        _groupManager->insertGroup(_pid, _gid);
    }

    UINT EasyParallelJob::newGroup()
    {
        return _groupManager->newGroup();
    }

    HRESULT EasyParallelJob::initializeTask(INOUT FrameTask* _task)
    {
        HRESULT hr = H_OK;

#ifdef LOG_EASY_PARALLEL_JOB
        {
            std::stringstream _buf;
            _buf << "Start initialize task : ["  << _task->getTaskId() << "] " << _task->getTaskName();
            _myFrame->getLogRecorder()->log(
                H_OK,
                bcmi_frame::JOB,
                __FUNCSIG__,
                (CSTRING)_buf.str().c_str()
                );
                _myFrame->getLogRecorder()->logFlush();
        }
#endif //#ifdef LOG_EASY_PARALLEL_JOB

        hr = _task->frameInitialize(); // frame do
        if (FAILED(hr))
        {
            //UNDO Log
            return hr;
        }
        hr = _task->initialize(); // user do
        if (FAILED(hr))
        {
            //undo Log
            return hr;
        }

#ifdef LOG_EASY_PARALLEL_JOB
        _myFrame->getLogRecorder()->log(
            H_OK,
            bcmi_frame::JOB,
            __FUNCSIG__,
            "Initialize done."
            );
        _myFrame->getLogRecorder()->logFlush();
#endif //#ifndef LOG_EASY_PARALLEL_JOB

        return hr;
    }

    HRESULT EasyParallelJob::finalizeTask(INOUT FrameTask* _task)
    {
        HRESULT hr = H_OK;

#ifdef LOG_EASY_PARALLEL_JOB
        {
            std::stringstream _buf;
            _buf << "Start finalize task : ["  << _task->getTaskId() << "] " << _task->getTaskName();
            _myFrame->getLogRecorder()->log(
                H_OK,
                bcmi_frame::JOB,
                __FUNCSIG__,
                (CSTRING)_buf.str().c_str()
            );
            _myFrame->getLogRecorder()->logFlush();
        }
#endif //#ifdef LOG_EASY_PARALLEL_JOB

        _task->finalize(); // user do
        if (FAILED(hr))
        {
            //UNDO Log
            return hr;
        }
        _task->frameFinalize(); // frame do
        if (FAILED(hr))
        {
            //UNDO Log
            return hr;
        }

#ifdef LOG_EASY_PARALLEL_JOB
        _myFrame->getLogRecorder()->log(
            H_OK,
            bcmi_frame::JOB,
            __FUNCSIG__,
            "Finalize done."
            );
        _myFrame->getLogRecorder()->logFlush();
#endif //#ifndef LOG_EASY_PARALLEL_JOB

        return hr;
    }

    HRESULT EasyParallelJob::executeTask(INOUT FrameTask* _task)
    {
        HRESULT hr = H_OK;

#ifdef LOG_EASY_PARALLEL_JOB
        {
            std::stringstream _buf;
            _buf << "Start execute task : ["  << _task->getTaskId() << "] " << _task->getTaskName();
            _myFrame->getLogRecorder()->log(
                H_OK,
                bcmi_frame::JOB,
                __FUNCSIG__,
                (CSTRING)_buf.str().c_str()
            );
            _myFrame->getLogRecorder()->logFlush();
        }
#endif //#ifdef LOG_EASY_PARALLEL_JOB

        if (bcmi_frame::MASTER_REMOTE_ID == _myFrame->getMPIProcessId())
        {
            hr = _task->masterHandler();
        }
        {
            hr = _task->execute();
        }
        if (FAILED(hr))
        {
            //UNDO Log
            return hr;
        }

#ifdef LOG_EASY_PARALLEL_JOB
        _myFrame->getLogRecorder()->log(
            H_OK,
            bcmi_frame::JOB,
            __FUNCSIG__,
            "Execute done."
            );
        _myFrame->getLogRecorder()->logFlush();
#endif //#ifndef LOG_EASY_PARALLEL_JOB

        return hr;
    }

    HRESULT EasyParallelJob::execute()
    {
        HRESULT hr = H_OK;

#ifdef LOG_EASY_PARALLEL_JOB
        _myFrame->getLogRecorder()->log(
            H_OK,
            bcmi_frame::JOB,
            __FUNCSIG__,
            "Start execute job."
            );
#endif //#ifdef LOG_EASY_PARALLEL_JOB

        if (_myFrame->getMPIProcessId() == bcmi_frame::MASTER_REMOTE_ID)
        {
            //UNDO Log
            hr = masterNodeExecute();
            if (FAILED(hr))
            {
                //UNDO Log
                return hr;
            }
        }
        else 
        {
            //UNDO Log
            hr = restNodeExecute();
            if (FAILED(hr))
            {
                //UNDO Log
                return hr;
            }
        }

#ifdef LOG_EASY_PARALLEL_JOB
        _myFrame->getLogRecorder()->log(
            H_OK,
            bcmi_frame::JOB,
            __FUNCSIG__,
            "Job execute done."
            );
#endif //#ifdef LOG_EASY_PARALLEL_JOB

        return hr;
    }

    HRESULT EasyParallelJob::restNodeExecute()
    {
        HRESULT hr = H_OK;

        EasyParallelTask *_initTask;
        TASK_MAP::iterator _taskIte = _taskMap.find(__FRAME_INI_TASK_NAME);
        if (_taskMap.end() == _taskIte
            || NULL == _taskIte->second)
        {
            //UNDO Log
            return H_E_FRAME_FIN_TASK_NOT_FOUND;
        }
        _initTask = (EasyParallelTask*)(_taskIte->second);
        hr = initializeTask(_initTask);
        if (FAILED(hr))
        {
            //UNDO Log
            return hr;
        }
        hr = executeTask(_initTask); //run initialize task
        if (FAILED(hr))
        {
            //UNDO Log
            return hr;
        }
        
        EasyParallelTask *_currentTask;
        _currentTask = _initTask;
        while (TRUE)
        {
            hr = genAndSendTaskRequest(_currentTask);
            if (FAILED(hr))
            {
                //UNDO Log
                return hr;
            }

            EasyParallelTask *_nextTask = NULL;
            hr = receiveAndGotoNextTask((FrameTask**)&_nextTask);
            if (FAILED(hr))
            {
                //UNDO Log
                return hr;
            }

            hr = finalizeTask(_currentTask);
            if (FAILED(hr))
            {
                //UNDO Log
                return hr;
            }

            _currentTask = _nextTask; // do next task


#ifdef DEBUG_EASY_PARALLEL_JOB
/*
#define STDOUT std::cout << "~~~~~~"
            STDOUT << _taskList.size() << std::endl;
            STDOUT << _currentTask << std::endl;
#undef STDOUT //#define STDOUT std::cout << "~~~~~~"
*/
#endif //#ifdef DEBUG_EASY_PARALLEL_JOB


            hr = initializeTask(_currentTask);
            if (bcmi_frame::FINALIZE == _currentTask->getTaskType())
            {
                //UNDO Log
                break;
            }

            hr = executeTask(_currentTask);
            if (FAILED(hr))
            {
                //UNDO Log
                return hr;
            }
        }

        EasyParallelTask *_finalizeTask = _currentTask;
        hr = executeTask(_finalizeTask); //run finalize task
        if (FAILED(hr))
        {
            //UNDO Log
            return hr;
        }
        hr = finalizeTask(_finalizeTask);
        if (FAILED(hr))
        {
            //UNDO Log
            return hr;
        }

        return hr;
    }

    HRESULT EasyParallelJob::genAndSendTaskRequest(IN FrameTask* _task)
    {
        HRESULT hr = H_OK;
        BYTE *_buf;
        newArray(&_buf, _requestBufferSize);
        UINT _curId = _task->getTaskId();
        memcpy(_buf, (BYTE*)(&_curId), sizeof(UINT));
        if (_taskMap.find(_task->nextTaskName()) == _taskMap.end())
        {
            //UNDO Log
            delete[] _buf;
            return H_E_NEXT_TASK_NOT_FOUND;
        }
        UINT _nextId = (_taskMap.find(_task->nextTaskName())->second)->getTaskId();
        memcpy((_buf + sizeof(UINT)), (BYTE*)(&_nextId), sizeof(UINT));
        for (UINT i = 0; i < _myFrame->getMPIProcessNumber(); ++i)
        {
            UINT _bid = sizeof(UINT) + sizeof(UINT) + i;
            if (_task->_writer.find(i) != _task->_writer.end())
            {
                _buf[_bid] = (BYTE)1;
            }
            else 
            {
                _buf[_bid] = (BYTE)0;
            }
        }

#ifdef LOG_EASY_PARALLEL_JOB
        {
            {
                std::stringstream _ibuf;
                _ibuf << "Start send the protocol of task : [" << _task->getTaskId() << "] " <<  _task->getTaskName();
                _myFrame->getLogRecorder()->log(
                    H_OK,
                    bcmi_frame::JOB,
                    __FUNCSIG__,
                    (CSTRING)_ibuf.str().c_str()
                    );
            }
            {
                std::stringstream _ibuf;
                _ibuf << "Next task id : " << _nextId;
                _myFrame->getLogRecorder()->log(
                    H_OK,
                    bcmi_frame::JOB,
                    __FUNCSIG__,
                    (CSTRING)_ibuf.str().c_str()
                    );
            }
            {
                std::stringstream _ibuf;
                _ibuf << "Send to process id :";
                for (UINT i = 0; i < _myFrame->getMPIProcessNumber(); ++i)
                {
                    if (_buf[sizeof(UINT) * 2 +i] == (BYTE)1)
                    {
                        _ibuf << " " << i;
                    }
                }
                _ibuf << " OVER";
                _myFrame->getLogRecorder()->log(
                    H_OK,
                    bcmi_frame::JOB,
                    __FUNCSIG__,
                    (CSTRING)_ibuf.str().c_str()
                    );
            }
            {
                std::stringstream _ibuf;
                _ibuf << "Send protocol buffer : [";
                for (UINT i = 0; i < _requestBufferSize; ++i)
                {
                    _ibuf << " " << std::hex << (INT)(_buf[i]);
                }
                _ibuf << " ]";
                _myFrame->getLogRecorder()->log(
                    H_OK,
                    bcmi_frame::JOB,
                    __FUNCSIG__,
                    (CSTRING)_ibuf.str().c_str()
                    );
            }
            _myFrame->getLogRecorder()->logFlush();
        }
#endif //#ifdef LOG_EASY_PARALLEL_JOB

        // MPI send protocol buffer
        MPI_Send(
            _buf,
            _requestBufferSize,
            MPI_BYTE,
            bcmi_frame::MASTER_REMOTE_ID,
            bcmi_frame::FRAME_PROTOCOL_TAG,
            MPI_COMM_WORLD);

#ifdef LOG_EASY_PARALLEL_JOB
        _myFrame->getLogRecorder()->log(
            H_OK,
            bcmi_frame::JOB,
            __FUNCSIG__,
            "Protocol send done."
            );
        _myFrame->getLogRecorder()->logFlush();
#endif //#ifdef LOG_EASY_PARALLEL_JOB

        delete[] _buf;
        return hr;
    }

    HRESULT EasyParallelJob::receiveAndGotoNextTask(OUT FrameTask* *_task)
    {
        HRESULT hr = H_OK;

#ifdef LOG_EASY_PARALLEL_JOB
        _myFrame->getLogRecorder()->log(
            H_OK,
            bcmi_frame::JOB,
            __FUNCSIG__,
            "Protocol receive."
            );
        _myFrame->getLogRecorder()->logFlush();
#endif //#ifdef LOG_EASY_PARALLEL_JOB

        BYTE *_buf;
        newArray(&_buf, _requestBufferSize);
        MPI_Status _status;
        MPI_Recv(
            _buf,
            _requestBufferSize,
            MPI_BYTE,
            bcmi_frame::MASTER_REMOTE_ID,
            bcmi_frame::FRAME_PROTOCOL_TAG,
            MPI_COMM_WORLD,
            &_status
            );
        //UNDO status analysis

#ifdef LOG_EASY_PARALLEL_JOB
        {
            std::stringstream _ibuf;
            _ibuf << "Receive done. Protocol buffer : [";
            for (UINT i = 0; i < _requestBufferSize; ++i)
            {
                _ibuf << " " << std::hex << (INT)(_buf[i]);
            }
            _ibuf << "]";
            _myFrame->getLogRecorder()->log(
                H_OK,
                bcmi_frame::JOB,
                __FUNCSIG__,
                (CSTRING)_ibuf.str().c_str()
                );
            _myFrame->getLogRecorder()->logFlush();
        }
#endif //#ifdef LOG_EASY_PARALLEL_JOB


        UINT _curId;
        memcpy((BYTE*)(&_curId), _buf, sizeof(UINT));
        //UNDO _curId check;

        UINT _nextId;
        memcpy((BYTE*)(&_nextId), (_buf + sizeof(UINT)), sizeof(UINT));
        if (_nextId < 0 || _nextId >= _taskList.size() || NULL == _taskList[_nextId])
        {
            //UNDO Log
            return H_E_NEXT_TASK_NOT_FOUND;
        }

        (*_task) = _taskList[_nextId];
        for (UINT i = 0; i < _myFrame->getMPIProcessNumber(); ++i)
        {
            UINT _bid = sizeof(UINT) + sizeof(UINT) + i;
            if ((BYTE)1 == _buf[_bid])
            {
                (*_task)->insertRecvId(i);
            }
        }


#ifdef LOG_EASY_PARALLEL_JOB
        {
            {
                std::stringstream _ibuf;
                _ibuf << "Protocol receive done. Task : [" << (*_task)->getTaskId() << "] " <<  (*_task)->getTaskName();
                _myFrame->getLogRecorder()->log(
                    H_OK,
                    bcmi_frame::JOB,
                    __FUNCSIG__,
                    (CSTRING)_ibuf.str().c_str()
                    );
            }
            {
                std::stringstream _ibuf;
                _ibuf << "Recv from process id :";
                for (UINT i = 0; i < _myFrame->getMPIProcessNumber(); ++i)
                {
                    if (_buf[sizeof(UINT) * 2 +i] == (BYTE)1)
                    {
                        _ibuf << " " << i;
                    }
                }
                _ibuf << " OVER";
                _myFrame->getLogRecorder()->log(
                    H_OK,
                    bcmi_frame::JOB,
                    __FUNCSIG__,
                    (CSTRING)_ibuf.str().c_str()
                    );
            }
            _myFrame->getLogRecorder()->logFlush();
        }
#endif //#ifdef LOG_EASY_PARALLEL_JOB

        delete[] _buf;
        return hr;
    }

    HRESULT EasyParallelJob::masterNodeExecute()
    {

#ifdef LOG_EASY_PARALLEL_JOB
        _myFrame->getLogRecorder()->log(
            H_OK,
            bcmi_frame::JOB,
            __FUNCSIG__,
            "Start master node execute."
            );
#endif //#ifdef LOG_EASY_PARALLEL_JOB

        HRESULT hr = H_OK;
        _processManager = new EasyParallelProcessManager(_groupManager);
        BYTE *_processRequestBuffer;
        MPI_Status _processStatus;

        newArray(&_processRequestBuffer, _requestBufferSize);

        while (TRUE)
        {
            MPI_Recv( // recv request buffer
                _processRequestBuffer,
                _requestBufferSize,
                MPI_BYTE,
                MPI_ANY_SOURCE,
                bcmi_frame::FRAME_PROTOCOL_TAG,
                MPI_COMM_WORLD,
                &_processStatus
                );
            //UNDO status analysis

#ifdef LOG_EASY_PARALLEL_JOB
            {
                std::stringstream _ibuf;
                _ibuf << "Recv protocol buffer from process " 
                    << _processStatus.MPI_SOURCE << " : [";
                for (UINT i = 0; i < _requestBufferSize; ++i)
                {
                    _ibuf << " " << std::hex << (INT)(_processRequestBuffer[i]);
                }
                _ibuf << "]";
                _myFrame->getLogRecorder()->log(
                    H_OK,
                    bcmi_frame::JOB,
                    __FUNCSIG__,
                    (CSTRING)_ibuf.str().c_str()
                    );
                _myFrame->getLogRecorder()->logFlush();
            }
#endif //#ifdef LOG_EASY_PARALLEL_JOB


            UINT _pid = _processStatus.MPI_SOURCE;
            UINT _gid = _groupManager->askGroup(_pid);

            _processManager->request(_pid);
            _processManager->parse(_pid, _processRequestBuffer);

#ifdef DEBUG_EASY_PARALLEL_JOB
#define STDCOUT std::cout << "......"
            STDCOUT << _pid << " "
                << _gid << " "
//                << _processManager->_groupManager->groupSize(_gid) << " "
//                << (_processManager->_groupManager->groupNumber()) << " "
//                << _processManager->_groupEmptyNumber[_gid] << std::endl;
                << std::endl;

#undef STDCOUT
#endif //#ifdef DEBUG_EASY_PARALLEL_JOB


            if (_processManager->allSendRequest(_gid))
            {

#ifdef LOG_EASY_PARALLEL_JOB
                {
                    std::stringstream _ibuf;
                    _ibuf << "All process in group " << _gid << " has received protocol.";
                    _myFrame->getLogRecorder()->log(
                        H_OK,
                        bcmi_frame::JOB,
                        __FUNCSIG__,
                        (CSTRING)_ibuf.str().c_str()
                        );
                    _myFrame->getLogRecorder()->logFlush();
                }
#endif //#ifdef LOG_EASY_PARALLEL_JOB

                hr = handleAndResponseGroup(_gid);
                if (FAILED(hr))
                {
                    //UNDO Log
                    return hr;
                }
            }

            if (_processManager->allProcessFinished() == TRUE) // all process has send finished(finalize task)
            { 
                //UNDO Log
                break;
            }
        }

        delete[] _processRequestBuffer;
        delete _processManager;
        return hr;
    }

    HRESULT EasyParallelJob::handleAndResponseGroup(IN UINT _gid)
    {
        HRESULT hr = H_OK;

        INT _ctid = _processManager->currentTask(_gid);
        UINT _currentTaskId;
        INT _ntid = _processManager->nextTask(_gid);

        UINT _nextTaskId;
        if (0 > _ctid // current task is not the same, go finalize
            || 0 > _ntid // next task is not the same, go finalize
            || _processManager->sendInGroup(_gid) == FALSE) // message out of group, go finalize
        {
            //UNDO Log
            _nextTaskId = (_taskMap.find(__FRAME_FIN_TASK_NAME)->second)->getTaskId();
            _currentTaskId = (_taskMap.find(__FRAME_INI_TASK_NAME)->second)->getTaskId();
        }
        else
        {
            _nextTaskId = (UINT)_ntid;
            _currentTaskId = (UINT)_ctid;
        }
        EasyParallelTask *_task = (EasyParallelTask*)_taskList[_nextTaskId];

        for (UINT i = 0; i < _myFrame->getMPIProcessNumber(); ++i)
        {
            if (_processManager->sendTo(i, bcmi_frame::MASTER_REMOTE_ID) == (BYTE)1)
            {
                _task->insertRecvId(i);
            }
        }

        hr = initializeTask(_task);
        if (FAILED(hr))
        {
            //UNDO Log
            return hr;
        }

        hr = executeTask(_task);
        if (FAILED(hr))
        {
            //UNDO Log
            return hr;
        }

        BYTE* _reponseBuffer;
        newArray(&_reponseBuffer, _requestBufferSize);
        UINT _headSize = sizeof(UINT) * 2;
        for (UINT i = 0; i < _myFrame->getMPIProcessNumber(); ++i)
        {
            if (_groupManager->askGroup(i) != _gid)
            {
                continue;
            }
            memcpy(_reponseBuffer, (BYTE*)(&_currentTaskId), sizeof(UINT));
            memcpy((_reponseBuffer + sizeof(UINT)), (BYTE*)(&_nextTaskId), sizeof(UINT));
            for (UINT j = 0; j < _myFrame->getMPIProcessNumber(); ++j)
            {
                _reponseBuffer[_headSize + j] = (BYTE)0;
                if (_groupManager->askGroup(j) != _gid)
                {
                    continue;
                }
                _reponseBuffer[_headSize + j] = _processManager->sendTo(i, j);
            }

#ifdef LOG_EASY_PARALLEL_JOB
            {
                std::stringstream _ibuf;
                _ibuf << "Send protocol buffer to process " << i << " : [";
                for (UINT i = 0; i < _requestBufferSize; ++i)
                {
                    _ibuf << " " << std::hex << (INT)(_reponseBuffer[i]);
                }
                _ibuf << "]";
                _myFrame->getLogRecorder()->log(
                    H_OK,
                    bcmi_frame::JOB,
                    __FUNCSIG__,
                    (CSTRING)_ibuf.str().c_str()
                    );
                _myFrame->getLogRecorder()->logFlush();
            }
#endif //#ifdef LOG_EASY_PARALLEL_JOB

            MPI_Send(
                _reponseBuffer,
                _requestBufferSize,
                MPI_BYTE,
                i,
                bcmi_frame::FRAME_PROTOCOL_TAG,
                MPI_COMM_WORLD
                );

#ifdef LOG_EASY_PARALLEL_JOB
            {
                _myFrame->getLogRecorder()->log(
                    H_OK,
                    bcmi_frame::JOB,
                    __FUNCSIG__,
                    "Send protocol done."
                    );
                _myFrame->getLogRecorder()->logFlush();
            }
#endif //#ifdef LOG_EASY_PARALLEL_JOB

            if (_task->getTaskType() != bcmi_frame::FINALIZE)
            {
                _processManager->reponseRequest(i);
            }
            else
            {
                _processManager->reponseFinished(i);
            }

        }
        delete[] _reponseBuffer;

        hr = finalizeTask(_task);
        if (FAILED(hr))
        {
            //UNDO Log
            return hr;
        }

        return hr;
    }

}; //namespace bcmi_frame